package r3.rpc;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import r3.common.R3Utils;
import r3.rpc.protocol.codec.RpcDecoder;
import r3.rpc.protocol.codec.RpcEncoder;
import r3.rpc.protocol.serialization.HessianSerialization;

/**
 * 客户端使用nio和服务器通讯
 * 每次通讯都创建一个新的RPCRequestLauncher对象
 *
 * @author zhoufn
 * @create 2017-12-25 14:02
 **/
@Slf4j
public class RpcRequestLauncher extends SimpleChannelInboundHandler<RpcResponse> {

    /**
     * 目标IP
     */
    private String host;

    /**
     * 目标端口
     */
    private int port;

    private RpcResponse rpcResponse;

    public RpcRequestLauncher(String host, int port) {
        this.host = host;
        this.port = port;
    }

    /**
     * 启动服务器发送请求
     *
     * @param rpcRequest
     * @return
     * @throws Exception
     */
    public RpcResponse launch(RpcRequest rpcRequest) throws Exception {
        if (rpcRequest.getRpcId() == null) {
            rpcRequest.setRpcId(R3Utils.randomUUID());
        }
        NioEventLoopGroup loopGroup = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(loopGroup)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new RpcDecoder(new HessianSerialization(), RpcResponse.class)).addLast(new RpcEncoder(new HessianSerialization()))
                                    .addLast(new SimpleChannelInboundHandler<RpcResponse>() {
                                        protected void channelRead0(ChannelHandlerContext ctx, RpcResponse msg) throws Exception {
                                            rpcResponse = msg;
                                        }
                                    });
                        }
                    });
            ChannelFuture future = bootstrap.connect(host, port).sync();
            log.debug("链接成功...");
            Channel channel = future.channel();
            channel.writeAndFlush(rpcRequest).sync();
            channel.closeFuture().sync();
        } catch (Exception e) {
            this.rpcResponse = new RpcResponse();
            rpcResponse.setRpcId(rpcRequest.getRpcId());
            rpcResponse.setException(e);
        } finally {
            loopGroup.shutdownGracefully();
        }
        return this.rpcResponse;
    }

    /**
     * <strong>Please keep in mind that this method will be renamed to
     * {@code messageReceived(ChannelHandlerContext, I)} in 5.0.</strong>
     * <p>
     *
     * @param ctx the {@link ChannelHandlerContext} which this {@link SimpleChannelInboundHandler}
     *            belongs to
     * @param msg the message to handle
     * @throws Exception is thrown if an error occurred
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcResponse msg) throws Exception {
        this.rpcResponse = msg;
    }

    public static void main(String[] args) throws Exception {
        for (int i = 0; i < 200; i++) {
            final int j = i;
            new Thread(() -> {
                try {
                    RpcRequestLauncher launcher = new RpcRequestLauncher("192.168.2.173", 20080);
                    RpcRequest request = new RpcRequest();
                    request.setRpcId("" + j);
                    request.setBeanId("sayByeWorker");
                    request.setMethodName("sayBye");
                    request.setParameterTypes(new Class[]{String.class});
                    request.setArguments(new Object[]{"zhoufn"});
                    RpcResponse response = launcher.launch(request);
                    System.out.println(j + ">>" + response.getValue().toString());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}
